interface MessageBus {
    publish(message: string, topic: string): void;
    subscribe(topic: string, channel: string[]): symbol;
    unsubscribe(id: symbol): void;
}

class MessageQueue implements MessageBus {
    private channels: Map<string, Map<symbol, string[]>>;
    private topics: Map<symbol, string>;
    constructor() {
        this.channels = new Map<string, Map<symbol, string[]>>();
        this.topics = new Map<symbol, string>();
    }
    private createTopic(topic: string): void {
        this.channels.set(topic, new Map<symbol, string[]>());
    }
    private closeTopic(topic: string): void {
        this.channels.delete(topic);
    }
    public publish(message: string, topic: string): void {
        if (!this.channels.has(topic)) {
            this.createTopic(topic);
        }
        for (const channel of this.channels.get(topic)!.values()) {
            channel.push(message);
        }
    }
    public subscribe(topic: string, channel: string[]): symbol {
        if (!this.channels.has(topic)) {
            this.createTopic(topic);
        }
        const id: symbol = Symbol();
        this.channels.get(topic)!.set(id, channel);
        return id;
    }
    public unsubscribe(id: symbol): void {
        const topic = this.topics.get(id);
        if (!topic) {
            return;
        }
        this.topics.delete(id);
        const channels: Map<symbol, string[]> = this.channels.get(topic) as Map<symbol, string[]>;
        channels.delete(id);
        if (channels.size === 0) {
            this.closeTopic(topic);
        }
    }
}

interface Producer {
    send(message: string, topic: string): void;
}

class Producer implements Producer {
    private mq: MessageBus;
    constructor(mq: MessageBus) {
        this.mq = mq
    }
    public send(message: string, topic: string): void {
        this.mq.publish(message, topic);
    }
}

interface Consumer {
    listen(topic: string): void;
    stopListen(topic: string): void;
    getMessage(): string;
}

class Consumer implements Consumer {
    private mq: MessageBus;
    private channel: string[];
    private topicIDs: Map<string, symbol>;
    constructor(mq: MessageBus) {
        this.mq = mq;
        this.channel = [];
        this.topicIDs = new Map<string, symbol>();
    }
    public listen(topic: string): void {
        const topicID: symbol = this.mq.subscribe(topic, this.channel);
        this.topicIDs.set(topic, topicID);
    }
    public stopListen(topic: string): void {
        const topicID = this.topicIDs.get(topic);
        if (!topicID) {
            return;
        }
        this.topicIDs.delete(topic);
        this.mq.unsubscribe(topicID);
    }
    public getMessage(): string {
        const message = this.channel.pop();
        if (!message) {
            return '';
        }
        return message;
    }
}

function main() {
    const mq: MessageBus = new MessageQueue();
    const producer: Producer = new Producer(mq);
    const consumer: Consumer = new Consumer(mq);
    const messages: string[] = ['python', 'java', 'golang', 'typescript', 'ruby'];
    const topics: string[] = ['test', 'demo'];
    let count: number = 0;
    const timer = setInterval(() => {
        const msg = messages.shift();
        if (!msg) {
            producer.send('<end>', 'test');
            clearInterval(timer);
            return;
        }
        const topic: string = topics[(count++) % 2]
        producer.send(msg, topic);
    }, 1000);
    consumer.listen('test');
    const timer2 = setInterval(() => {
        let msg: string = consumer.getMessage();
        if (msg == '<end>') {
            consumer.stopListen('test');
            clearInterval(timer2);
            return;
        }
        if (msg != '') {
            console.log(msg);
        }
    }, 100);
}

main();
